#include "EventLoop.h"
#include "Logger.h"
#include "Channel.h"

#include <cstring>
#include <sys/eventfd.h>
#include <memory>

__thread EventLoop *t_loopInThisThread = nullptr;

// epoll_wait超时返回的时间
const int kPollTimeMs = 10000;

int createEventfd()
{
    int evtfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    if (evtfd < 0)
    {
        LOG_FATAL("eventfd error: %d: %s", errno, strerror(errno));
    }
    return evtfd;
}

EventLoop::EventLoop()
    : looping_(false)
    , quit_(false)
    , threadId_(CurrentThread::tid())
    , poller_(Poller::newDefaultPoller(this))
    , wakeupFd_(createEventfd())
    , wakeupChannel_(new Channel(this, wakeupFd_))
    , callingPendingFunctors_(false)
    , timerWheel_(this) 
{
    LOG_INFO("EventLoop created %p, in thread %d", this, threadId_);
    if (t_loopInThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d", t_loopInThisThread, threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }

    // 对wakeupFd注册 被唤醒后的读操作
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
    // 设置wakeupFd读事件被关心，并托管给poller
    wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

// 开启事件循环
void EventLoop::loop()
{
    looping_ = true;
    quit_ = false;

    LOG_DEBUG("EventLoop %p start looping", this);

    while (!quit_)
    {
        activeChannels_.clear(); // 准备接收发生事件的channel，先清空
        // 这里监听两类fd，一类是用户fd，一类是自己的wakeupfd
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);

        for (Channel *channel : activeChannels_)
        {
            // LOG_DEBUG("发生事件: %d",channel->revents());
            //  EventLoop的poller监听到就绪的channel，上报给EventLoop，然后EVentLoop通知相应的channel处理事件
            channel->handleEvent(pollReturnTime_);
        }

        // 对新连接进行处理，即对新连接产生的fd对应的新channel托管给自己的（subloop）的poller
        doPendingFunctors();
    }

    LOG_INFO("EventLoop %p stop looping", this);
    looping_ = false;
}

// 退出事件循环：1.loop在自己的线程中调用自己的quit  2.loop在线程中调用其他loop的quit
void EventLoop::quit()
{
    quit_ = true;

    if (!isInLoopThread()) // 2这种情况，其他线程可能正阻塞了，无法往下执行再去进行while(!quit_)的判断，故对其唤醒
    {
        wakeup();
    }
}

// 工作时序：
//(runInLoop() -> quueInLoop())/queueInLoop() -> wakeup() -> poll() -> handleRead() -> doPendingFunctors()
void EventLoop::runInLoop(Functor cb)
{
    if (isInLoopThread()) // 在当前loop中执行cb
    {
        cb();
    }
    else // 在非当前loop的线程调用当前loop的runLoop
    {
        queueInLoop(cb);
    }
}

/**
 * 把cb放入loop队列中，唤醒loop所在线程执行cb
 * 在非当前loop的线程调用当前loop的runLoop，会执行queueInLoop
 * 先把cb放入当前loop队列，然后唤醒当前loop线程去执行队列中的cb
 */
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    /**
     * 唤醒当前loop线程
     * || callingPendingFunctors_ 当loop正在执行doPendingFunctors时，拷贝完后，mainloop又派发了新的cb，
     * doPendingFunctors结束后，再循环可能又在poll处阻塞。所以页需要wakeup()，发送数据唤醒一下
     */
    if (!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup();
    }
}

// 通过往wakeupfd中写一个数据，而使其读事件发生，从而唤醒当前loop的线程
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8", n);
    }
}

// 调用poller的接口
void EventLoop::updateChannel(Channel *channel)
{
    poller_->updateChannel(channel);
}
void EventLoop::removeChannel(Channel *channel)
{
    poller_->removeChannel(channel);
}
bool EventLoop::hasChannel(Channel *channel)
{
    return poller_->hasChannel(channel);
}

// wakupchannel读事件的处理函数
void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof(one));
    if (n != sizeof(one))
    {
        LOG_ERROR("EventLoop::handleRead() reads %lu bytes instead of 8", n);
    }
}

// 执行回调
void EventLoop::doPendingFunctors()
{
    /**
     * 如果是从pendingFunctors中拿一个，删掉，执行，再重复，那么锁会一直被持有，直到pendingFunctors为空
     * 那么mainloop往pendingFunctors中派发时竞争不到锁，可能会阻塞在queueInLoop处
     *
     * swap，取出来后锁就释放了，可以循环去执行cb，而又不阻塞mianloop的派发，实现这俩的并发
     */
    std::vector<Functor> functors;
    callingPendingFunctors_ = true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFunctors_);
    }

    for (const auto &functor : functors)
    {
        functor();
    }

    callingPendingFunctors_ = false;
}

void EventLoop::timerAdd(std::string name, uint32_t delay, const TaskFunc &cb)
{
    timerWheel_.timerAdd(name, delay, cb);
}
void EventLoop::timerRefresh(std::string name)
{
    timerWheel_.timerRefresh(name);
}
void EventLoop::timerCancel(std::string name)
{
    timerWheel_.timerCancel(name);
}
bool EventLoop::hasTimer(std::string name)
{
    return timerWheel_.hasTimer(name);
}